package com.fand.stater.listener;

import java.util.Map;
import java.util.Objects;

import javax.annotation.Resource;

import cn.hutool.core.annotation.AnnotationUtil;
import com.fand.stater.annotation.MqSubscribe;
import com.fand.stater.autoconfigure.properties.ConsumerConfig;
import com.fand.stater.process.RocketMQProcessor;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.MessageListener;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.springframework.context.ApplicationListener;
import org.springframework.context.event.ContextRefreshedEvent;

/**
 * @Author: zhangjiabin
 * @Date: 2021/5/27
 * @Version 1.0
 */
@Slf4j
public class MQConsumerLoadEvent implements ApplicationListener<ContextRefreshedEvent> {

    @Resource
    private ConsumerConfig consumerConfig;

    @Override
    public void onApplicationEvent(ContextRefreshedEvent contextRefreshedEvent) {
        Map<String,Object> mapListener = RocketMQProcessor.mapListener;
        if(mapListener.size() < 1) return;
        mapListener.forEach((name,bean)->{
            DefaultMQPushConsumer defaultMqPushConsumer = new DefaultMQPushConsumer(name);
            defaultMqPushConsumer.setNamesrvAddr(consumerConfig.getNamesrvAddr());
            defaultMqPushConsumer.setVipChannelEnabled(false);
            defaultMqPushConsumer.setConsumeThreadMin(consumerConfig.getConsumeThreadMin());
            defaultMqPushConsumer.setConsumeThreadMax(consumerConfig.getConsumeThreadMax());
            defaultMqPushConsumer.setConsumeMessageBatchMaxSize(consumerConfig.getConsumeMessageBatchMaxSize());
            defaultMqPushConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
            MqSubscribe annotation = AnnotationUtil.getAnnotation(bean.getClass(), MqSubscribe.class);
            try {
                defaultMqPushConsumer.subscribe(Objects.nonNull(annotation.topic())?annotation.topic():"",Objects.nonNull(annotation.tag())?annotation.tag():"");
                defaultMqPushConsumer.registerMessageListener((MessageListener)bean);
                defaultMqPushConsumer.start();
            } catch (MQClientException e) {
                e.printStackTrace();
            }
        });
    }

}
